# -*- coding: utf-8 -*-
"""
流式 ASR WebSocket 服务器，可选配唤醒词模式。
"""

# ============================================================
# 🛠 依赖库
# ============================================================
from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Request, HTTPException
from fastapi.exceptions import RequestValidationError
from fastapi.responses import JSONResponse, HTMLResponse
from fastapi.middleware.cors import CORSMiddleware
from starlette.status import HTTP_422_UNPROCESSABLE_ENTITY
from pydantic_settings import BaseSettings,SettingsConfigDict
from pydantic import BaseModel, Field
from funasr import AutoModel # FunASR 自动模型加载
from modelscope.pipelines import pipeline # ModelScope Pipeline 接口
from modelscope.utils.constant import Tasks # ModelScope 任务常量
from loguru import logger # 日志库

import numpy as np # NumPy 用于数值计算
import soundfile as sf # SoundFile 用于读写音频文件
import argparse # Argparse 用于命令行参数解析
import uvicorn # Uvicorn 是一个 ASGI 服务器
from urllib.parse import parse_qs # 用于解析 URL 查询字符串
import os # OS 模块用于与操作系统交互
import re # Re 模块用于正则表达式操作
import sys # Sys 模块用于与 Python 解释器交互
import json # JSON 模块用于处理 JSON 数据
import traceback # Traceback 模块用于获取异常堆栈信息
import time # Time 模块用于时间相关操作
from typing import List # Typing 模块用于类型提示
from pypinyin import lazy_pinyin # Pypinyin 用于汉字转拼音

# ============================================================
# 📜 日志
# ============================================================
logger.remove() # 移除默认的 logger handler
# 定义日志格式，包含时间、级别、文件名、行号和消息
log_format = "{time:YYYY-MM-DD HH:mm:ss} [{level}] {file}:{line} - {message}"
# 添加标准输出 handler，级别为 DEBUG，过滤掉 ERROR 及以上级别的日志
logger.add(sys.stdout, format=log_format, level="DEBUG", filter=lambda r: r["level"].no < 40)
# 添加标准错误输出 handler，级别为 ERROR，仅记录 ERROR 及以上级别的日志
logger.add(sys.stderr, format=log_format, level="ERROR", filter=lambda r: r["level"].no >= 40)

# ============================================================
# ⚙️ 配置
# ============================================================
class Config(BaseSettings):
    # ───────── 语音处理管线配置 ─────────
    sv_thr: float = Field(0.3, description="说话人验证阈值")
    chunk_size_ms: int = Field(300, description="VAD/ASR 分块大小 (毫秒)")
    sample_rate: int = Field(16000, description="采样率 (Hz)")
    bit_depth: int = Field(16, description="位深")
    channels: int = Field(1, description="声道数")
    avg_logprob_thr: float = Field(-0.25, description="平均对数概率阈值 (ASR)")
    max_end_silence_time :int = Field(2000, description="最大末尾静音时长 (毫秒)")

    # ───────── 唤醒词配置 ─────────
    wake_words: List[str] = Field(["你好助手", "语音助手"], description="唤醒词列表")
    wake_timeout: int = Field(120, description="ASR 无语音超时 (秒)")

    model_config = SettingsConfigDict(
        env_file=".env",          # 指定 env 文件
        env_file_encoding="utf-8" # 如果有中文最好显式写编码
    )

config = Config() # 实例化配置对象

# 预计算唤醒词的拼音（小写且无空格），用于快速匹配
# 例如 "你好助手" -> "nihaozhushou"
WAKE_PINYINS = {"".join(lazy_pinyin(w)).lower() for w in config.wake_words}

# ============================================================
# 😊 Emoji / 标签映射表（原样保留，但添加中文注释）
# ============================================================

# 情感标签到 Emoji 的映射
emo_dict = {
	"<|HAPPY|>": "😊",    # 开心
	"<|SAD|>": "😔",      # 伤心
	"<|ANGRY|>": "😡",    # 生气
	"<|NEUTRAL|>": "",    # 中性 (无 Emoji)
	"<|FEARFUL|>": "😰",   # 害怕
	"<|DISGUSTED|>": "🤢", # 厌恶
	"<|SURPRISED|>": "😮", # 惊讶
}

# 事件标签到 Emoji 或符号的映射
event_dict = {
	"<|BGM|>": "🎼",      # 背景音乐
	"<|Speech|>": "",     # 语音 (无特定 Emoji)
	"<|Applause|>": "👏", # 鼓掌
	"<|Laughter|>": "😀", # 笑声
	"<|Cry|>": "😭",      # 哭泣
	"<|Sneeze|>": "🤧",   # 打喷嚏
	"<|Breath|>": "",     # 呼吸 (无特定 Emoji)
	"<|Cough|>": "🤧",    # 咳嗽 (使用打喷嚏的 Emoji，可能需要调整)
}

# 综合性的标签到 Emoji/符号的映射表，用于 `format_str` 函数
emoji_dict = {
	"<|nospeech|><|Event_UNK|>": "❓", # 无语音且事件未知
	"<|zh|>": "",          # 中文语种标签 (替换为空)
	"<|en|>": "",          # 英文语种标签 (替换为空)
	"<|yue|>": "",         # 粤语语种标签 (替换为空)
	"<|ja|>": "",          # 日语语种标签 (替换为空)
	"<|ko|>": "",          # 韩语语种标签 (替换为空)
	"<|nospeech|>": "",    # 无语音标签 (替换为空)
	"<|HAPPY|>": "😊",     # 开心
	"<|SAD|>": "😔",       # 伤心
	"<|ANGRY|>": "😡",     # 生气
	"<|NEUTRAL|>": "",     # 中性
	"<|BGM|>": "🎼",       # 背景音乐
	"<|Speech|>": "",      # 语音
	"<|Applause|>": "👏",  # 鼓掌
	"<|Laughter|>": "😀",  # 笑声
	"<|FEARFUL|>": "😰",    # 害怕
	"<|DISGUSTED|>": "🤢",  # 厌恶
	"<|SURPRISED|>": "😮",  # 惊讶
	"<|Cry|>": "😭",       # 哭泣
	"<|EMO_UNKNOWN|>": "", # 未知情感 (替换为空)
	"<|Sneeze|>": "🤧",    # 打喷嚏
	"<|Breath|>": "",      # 呼吸
	"<|Cough|>": "😷",     # 咳嗽 (这里用戴口罩的 Emoji)
	"<|Sing|>": "",        # 唱歌 (替换为空)
	"<|Speech_Noise|>": "",# 带噪声的语音 (替换为空)
	"<|withitn|>": "",     # ITN (Inverse Text Normalization) 相关标签 (替换为空)
	"<|woitn|>": "",       # 无 ITN 相关标签 (替换为空)
	"<|GBG|>": "",         # 垃圾信息/无关语音 (替换为空)
	"<|Event_UNK|>": "",   # 未知事件 (替换为空)
}

# 语种标签统一替换为 "<|lang|>" 的映射
lang_dict =  {
    "<|zh|>": "<|lang|>",
    "<|en|>": "<|lang|>",
    "<|yue|>": "<|lang|>",
    "<|ja|>": "<|lang|>",
    "<|ko|>": "<|lang|>",
    "<|nospeech|>": "<|lang|>", # 无语音也视为一种语言标记处理单元
}

# 情感相关的 Emoji 集合
emo_set = {"😊", "😔", "😡", "😰", "🤢", "😮"}
# 事件相关的 Emoji/符号集合
event_set = {"🎼", "👏", "😀", "😭", "🤧", "😷",}

# 格式化字符串函数：简单地将 emoji_dict 中的标签替换为对应的 Emoji 或空字符串
def format_str(s):
	for sptk in emoji_dict: # 遍历 emoji_dict 中的所有特殊标签
		s = s.replace(sptk, emoji_dict[sptk]) # 将标签替换为其映射值
	return s


# 格式化字符串函数版本2：
# 1. 统计输入字符串中各特殊标签的出现次数。
# 2. 移除所有特殊标签，保留纯文本。
# 3. 根据统计结果，确定主要情感（出现次数最多的情感标签），并在文本末尾附加对应 Emoji。
# 4. 如果存在事件标签，将对应事件符号加在文本开头。
# 5. 清理多余的空格。
def format_str_v2(s):
	sptk_dict = {} # 用于存储各特殊标签的计数字典
	for sptk in emoji_dict: # 遍历所有可能的特殊标签
		sptk_dict[sptk] = s.count(sptk) # 统计该标签在字符串 s 中出现的次数
		s = s.replace(sptk, "") # 从字符串 s 中移除该标签
	
	emo = "<|NEUTRAL|>" # 默认情感为中性
	for e in emo_dict: # 遍历 emo_dict 中的情感标签
		if sptk_dict[e] > sptk_dict[emo]: # 如果当前情感标签 e 的计数大于当前主要情感 emo 的计数
			emo = e # 更新主要情感为 e
	
	for e in event_dict: # 遍历 event_dict 中的事件标签
		if sptk_dict[e] > 0: # 如果事件标签 e 出现过
			s = event_dict[e] + s # 将事件对应的符号加在字符串 s 的开头
	
	s = s + emo_dict[emo] # 将主要情感对应的 Emoji 加在字符串 s 的末尾

	# 清理 Emoji 周围可能产生的多余空格
	for emoji in emo_set.union(event_set): # 遍历所有情感和事件 Emoji
		s = s.replace(" " + emoji, emoji) # 移除 Emoji 前的空格
		s = s.replace(emoji + " ", emoji) # 移除 Emoji 后的空格
	return s.strip() # 返回处理后并去除首尾空格的字符串

# 格式化字符串函数版本3：
# 这是一个更复杂的格式化逻辑，主要处理包含语种标签 "<|lang|>" 的情况。
# 它首先将所有语种标签（如 <|zh|>）替换为统一的 "<|lang|>"。
# 然后按 "<|lang|>" 分割字符串，对每个子串应用 format_str_v2 进行格式化。
def format_str_v3(s):
	# 辅助函数：获取字符串末尾的情感 Emoji (如果存在)
	def get_emo(s_part):
		return s_part[-1] if len(s_part) > 0 and s_part[-1] in emo_set else None
	# 辅助函数：获取字符串开头的事件符号 (如果存在)
	def get_event(s_part):
		return s_part[0] if len(s_part) > 0 and s_part[0] in event_set else None

	# 特殊处理 "<|nospeech|><|Event_UNK|>" 为 "❓"
	s = s.replace("<|nospeech|><|Event_UNK|>", "❓")
	# 将所有具体的语种标签替换为统一的 "<|lang|>" 标记
	for lang_tag in lang_dict:
		s = s.replace(lang_tag, "<|lang|>")
	
	# 按 "<|lang|>" 分割字符串，并对每个子串应用 format_str_v2
	s_list = [format_str_v2(s_i).strip(" ") for s_i in s.split("<|lang|>")]
	
	new_s = "" # 初始化最终结果字符串
	new_s = " " + s_list[0] # 第一个子串前面加个空格（后续会 strip）
	cur_ent_event = get_event(new_s.strip()) # 获取第一个子串的事件符号

	# 遍历剩余的子串进行合并
	for i in range(1, len(s_list)):
		if len(s_list[i]) == 0: # 跳过空子串
			continue
		
		current_segment_event = get_event(s_list[i])
		# 如果当前子串的事件符号与前一个合并后的事件符号相同，并且不是 None
		if current_segment_event == cur_ent_event and current_segment_event is not None:
			s_list[i] = s_list[i][1:] # 移除当前子串开头的事件符号，避免重复
		
		cur_ent_event = get_event(s_list[i]) # 更新当前事件符号为处理后（可能已移除）的当前子串的事件符号
		
		current_segment_emo = get_emo(s_list[i])
		# 如果当前子串的情感 Emoji 与 new_s 末尾的情感 Emoji 相同，并且不是 None
		if current_segment_emo is not None and current_segment_emo == get_emo(new_s.strip()):
			new_s = new_s.strip()[:-1] # 移除 new_s 末尾的情感 Emoji，避免重复
		
		new_s += s_list[i].strip().lstrip() # 合并当前子串（去除首尾及左侧多余空格）
	
	new_s = new_s.replace("The.", " ").replace("Sure.", " ").replace("Yeah.", " ")
	return new_s.strip() # 返回最终处理并去除首尾空格的字符串

# 检查字符串是否包含中文字符、英文字母或数字
def contains_chinese_english_number(s: str) -> bool:
    return bool(re.search(r'[\u4e00-\u9fffA-Za-z0-9]', s)) # 使用正则表达式匹配

# ============================================================
# 📦 模型加载
# ============================================================
logger.info("开始加载模型… (首次可能耗时较久)")

# 说话人验证模型 pipeline 初始化
sv_pipeline = pipeline(
    task=Tasks.speaker_verification, # 指定任务为说话人验证
    model="iic/speech_eres2net_large_sv_zh-cn_3dspeaker_16k", # 模型ID
    model_revision="v1.0.0", # 模型版本
)

# 流式 ASR (自动语音识别) 模型初始化
model_asr = AutoModel(
    model="iic/SenseVoiceSmall", # 模型ID或路径
    trust_remote_code=True, # 信任远程代码（如果模型定义在外部 Python 文件中）
    remote_code="./model.py", # 远程代码文件路径 (model.py 应该与此脚本在同一目录或指定正确路径)
    device="cuda:0", # 使用 CUDA 设备 0 (GPU)
    disable_update=True, # 禁止自动更新模型
)

# VAD (语音活动检测) 模型初始化
model_vad = AutoModel(
    model="fsmn-vad", # VAD 模型ID
    model_revision="v2.0.4", # 模型版本
    disable_pbar=True, # 禁用进度条显示
    max_end_silence_time=config.max_end_silence_time, # 最大末尾静音时长 (毫秒)
    disable_update=True, # 禁止自动更新模型
)

logger.info("模型加载完成！")

# ============================================================
# 🔈 说话人注册
# ============================================================
# 注册说话人的音频文件列表
reg_spks_files = [
    "speaker/speaker1_a_cn_16k.wav", # 示例说话人音频文件路径
]

# 初始化注册说话人数据
def reg_spk_init(files):
    reg_spk = {} # 存储注册说话人信息的字典
    for f in files:
        try:
            data, sr = sf.read(f, dtype="float32") # 读取音频文件，数据类型为 float32
            k, _ = os.path.splitext(os.path.basename(f)) # 从文件名中提取说话人标识 (如 "speaker1_a_cn_16k")
            reg_spk[k] = {"data": data, "sr": sr} # 存储音频数据和采样率
            logger.info(f"已注册说话人: {k}，音频长度: {len(data)/sr:.2f} 秒")
        except Exception as e:
            logger.error(f"注册说话人文件 {f} 失败: {e}")
    return reg_spk

reg_spks = reg_spk_init(reg_spks_files) # 执行说话人注册

# ============================================================
# 🔧 辅助: ASR 包装
# ============================================================

# ASR (自动语音识别) 函数封装
def asr(audio: np.ndarray, lang: str, cache: dict, use_itn: bool = False):
    start_time = time.time() # 记录开始时间
    # 调用 FunASR 模型的 generate 方法进行识别
    result = model_asr.generate(
        input=audio, # 输入音频 NumPy 数组
        cache=cache, # ASR 内部缓存，用于流式识别
        language=lang.strip(), # 识别语言 (去除首尾空格)
        use_itn=use_itn, # 是否使用逆文本归一化 (ITN)
        batch_size_s=60, # 推理时的批处理大小 (秒)
    )
    logger.debug(f"ASR 耗时: {(time.time() - start_time) * 1000:.2f} 毫秒") # 记录并打印 ASR 耗时
    return result

# ============================================================
# 🚀 FastAPI
# ============================================================
app = FastAPI() # 创建 FastAPI 应用实例

# 添加 CORS (跨源资源共享) 中间件，允许所有来源、凭证、方法和头部
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# ============================================================
# 📦 数据模型 / 异常处理
# ============================================================
# 定义响应体的数据模型
#class TranscriptionResponse(BaseModel):
#    code: int # 状态码
#    info: str # 状态信息
#    data: str | None = None # 返回数据，可以是字符串或 None
#    event: str | None = None # 事件类型，可以是字符串或 None
from typing import Union, Optional

class TranscriptionResponse(BaseModel):
    code: int  # 状态码
    info: str  # 状态信息
    data: Optional[str] = None  # 返回数据，可以是字符串或 None
    event: Optional[str] = None  # 事件类型，可以是字符串或 None

# 自定义全局异常处理器
@app.exception_handler(Exception)
async def custom_exception_handler(request: Request, exc: Exception):
    logger.error("捕获到未处理异常：", exc_info=True) # 记录异常信息，包括堆栈
    if isinstance(exc, HTTPException): # 如果是 FastAPI 的 HTTPException
        status_code, message = exc.status_code, exc.detail
    elif isinstance(exc, RequestValidationError): # 如果是 Pydantic 的请求校验错误
        status_code, message = HTTP_422_UNPROCESSABLE_ENTITY, "参数校验失败: " + str(exc.errors())
    else: # 其他未知异常
        status_code, message = 500, "服务器内部错误: " + str(exc)
    # 返回统一格式的 JSON 响应
    return JSONResponse(status_code=status_code, content=TranscriptionResponse(code=status_code, info=message).model_dump())

# ============================================================
# 🧩 核心: AudioStreamHandler
# ============================================================
class AudioStreamHandler:
    """处理单个 WebSocket 音频流，可选配唤醒词模式。"""

    MODE_WAKE = "wake" # 唤醒模式常量
    MODE_ASR = "asr"   # ASR 模式常量

    def __init__(self, websocket: WebSocket, *, sv: bool = False, lang: str = "auto", wake: bool = True):
        self.ws = websocket # WebSocket 连接对象
        self.sv = sv # 是否启用说话人验证 (Speaker Verification)
        self.lang = lang # ASR 识别语言
        self.enable_wake = wake # 是否启用唤醒词模式
        self.mode = self.MODE_WAKE if self.enable_wake else self.MODE_ASR # 当前运行模式

        # ───────── 运行时状态 ─────────
        self._running = False # 标记处理器是否正在运行
        self.buffer = b"" # 接收到的原始字节数据缓冲区
        # 每个音频块的大小 (样本数)，根据配置的 chunk_size_ms 和 sample_rate 计算
        self.chunk_size = int(config.chunk_size_ms * config.sample_rate / 1000) 
        self.audio_buffer = np.array([], dtype=np.float32) # 存储转换后的 float32 音频数据
        self.audio_vad = np.array([], dtype=np.float32) # 用于 VAD 处理的音频数据累积

        # 缓存
        self.cache_vad: dict = {} # VAD 模型的内部缓存
        self.cache_asr: dict = {} # ASR 模型的内部缓存
        self.last_vad_beg = -1 # 上一个 VAD 检测到的语音段开始时间 (毫秒)
        self.last_vad_end = -1 # 上一个 VAD 检测到的语音段结束时间 (毫秒)
        self.offset = 0 # VAD 时间戳偏移量，用于处理流式音频
        self.has_sent_begin = False # 是否已发送 onSentenceBegin 事件
        self.hit = False  # 说话人是否已验证通过

        # 唤醒词辅助变量
        self.last_active_time = time.monotonic() # 上次检测到活动（语音结束或唤醒词）的时间戳

    # ========================================================
    # 🏁 生命周期
    # ========================================================
    async def start(self):
        self._running = True # 设置运行状态为 True
        await self.ws.accept() # 接受 WebSocket 连接
        logger.info(f"WebSocket 已连接，客户端: {self.ws.client.host}:{self.ws.client.port}，当前模式: {self.mode}")

        # 连接后如果启用语音唤醒 ➔ 发送可用唤醒词列表
        if self.enable_wake:
            await self._send_event("onWakeWords", data=json.dumps(config.wake_words, ensure_ascii=False))

        try:
            while self._running: # 循环接收和处理音频数据
                data = await self.ws.receive_bytes() # 异步接收字节数据
                if not data: # 如果收到空数据 (可能表示连接关闭的信号)
                    logger.info("收到空字节数据，可能客户端正在关闭连接。")
                    break 
                self.buffer += data # 将接收到的数据追加到缓冲区
                
                # 确保缓冲区中至少有2个字节 (一个 int16 样本)
                if len(self.buffer) < 2:
                    continue
                
                # 计算可以转换为 int16 样本的字节长度 (必须是2的倍数)
                int16_len = len(self.buffer) - (len(self.buffer) % 2)
                # 将字节数据转换为 float32 NumPy 数组并归一化到 [-1, 1]
                new_audio = np.frombuffer(self.buffer[:int16_len], dtype=np.int16).astype(np.float32) / 32767.0
                self.audio_buffer = np.append(self.audio_buffer, new_audio)
                self.buffer = self.buffer[int16_len:] # 更新缓冲区，移除已处理的字节

                # 处理所有完整的音频块
                while len(self.audio_buffer) >= self.chunk_size:
                    await self._process_chunk()

        except WebSocketDisconnect:
            logger.info(f"客户端 {self.ws.client.host}:{self.ws.client.port} 断开连接")
        except Exception as e:
            logger.error(f"运行 AudioStreamHandler 时发生异常: {e}\n{traceback.format_exc()}")
            # 尝试安全关闭 WebSocket
            try:
                await self.ws.close(code=1011, reason="服务器内部错误") # 1011: Internal Error
            except RuntimeError: # 可能 WebSocket 已经关闭
                logger.warning("尝试关闭 WebSocket 时发生 RuntimeError (可能已关闭)")
        finally:
            await self.stop() # 确保资源被释放

    async def stop(self):
        if not self._running: # 如果已经停止，则不执行任何操作
            return
        self._running = False # 设置运行状态为 False
        # 清理音频缓冲区和缓存
        self.audio_buffer = np.array([], dtype=np.float32)
        self.audio_vad = np.array([], dtype=np.float32)
        self.cache_vad.clear()
        self.cache_asr.clear()
        logger.info(f"AudioStreamHandler 已停止，资源已释放 (客户端: {self.ws.client.host}:{self.ws.client.port})")

    # ========================================================
    # 🔀 模式切换
    # ========================================================
    def _switch_to_wake(self):
        if not self.enable_wake: # 如果未启用唤醒模式，则不执行切换
            return
        logger.info("🔄 切换到唤醒模式")
        self.mode = self.MODE_WAKE
        # 清理 ASR 相关状态，VAD 状态也重置以开始新的检测周期
        self.cache_asr.clear()
        self.cache_vad.clear() # 重置 VAD 缓存，以便从头开始检测
        self.audio_vad = np.array([], dtype=np.float32) # 清空 VAD 音频累积
        self.last_vad_beg = self.last_vad_end = -1 # 重置 VAD 时间戳
        self.offset = 0 # 重置 VAD 偏移
        self.has_sent_begin = False # 重置句子开始标记
        self.last_active_time = time.monotonic() # 重置活动时间戳

    def _switch_to_asr(self):
        logger.info("🔄 进入 ASR 模式")
        self.mode = self.MODE_ASR
        self.cache_asr.clear() # 清理 ASR 缓存，准备新的识别会话
        self.last_active_time = time.monotonic() # 更新活动时间戳
        # VAD 缓存和状态通常保留，以便当前正在进行的语音段可以继续被 ASR 处理
        # 如果需要完全重新开始，则也应清理 VAD 相关状态，但当前逻辑是连续的
        self.has_sent_begin = False # 重置，因为新的 ASR 会话开始了

    # ========================================================
    # 📤 事件发送
    # ========================================================
    async def _send_event(self, event: str, data: str = "", info: str = ""):
        resp = TranscriptionResponse(code=0, info=info, data=data, event=event)
        try:
            await self.ws.send_json(resp.model_dump()) # 发送 JSON 格式的事件数据
            logger.debug(f"➡️  发送事件 {event}: 数据='{data}', 信息='{info}'")
        except Exception as e:
            logger.error(f"发送事件 {event} 失败: {e}")
            # 如果发送失败，可能需要处理连接已断开的情况
            self._running = False # 标记为停止运行，以便主循环退出

    # ========================================================
    # 🧠 处理单个 chunk
    # ========================================================
    async def _process_chunk(self):
        chunk = self.audio_buffer[: self.chunk_size] # 取出一个音频块
        self.audio_buffer = self.audio_buffer[self.chunk_size :] # 更新 audio_buffer
        self.audio_vad = np.append(self.audio_vad, chunk) # 将块追加到 VAD 处理的音频流中

        # ─── VAD (语音活动检测) ───
        # 使用 VAD 模型处理当前音频块，is_final=False 表示流式处理
        vad_res = model_vad.generate(input=chunk, cache=self.cache_vad, is_final=False, chunk_size=config.chunk_size_ms)
        
        if vad_res and len(vad_res[0]["value"]): # 如果 VAD 检测到语音段信息
            vad_segments = vad_res[0]["value"] # 获取 VAD 分段结果
            for segment in vad_segments:
                # segment[0] 是语音开始时间 (ms), segment[1] 是语音结束时间 (ms)
                # -1 表示尚未确定
                if segment[0] > -1: # 如果检测到语音开始
                    self.last_vad_beg = segment[0]
                if segment[1] > -1: # 如果检测到语音结束
                    self.last_vad_end = segment[1]

                # 如果检测到语音开始 (last_vad_beg > -1)，且尚未检测到结束 (last_vad_end == -1)
                # 并且尚未发送 onSentenceBegin 事件，并且当前处于 ASR 模式
                if self.last_vad_beg > -1 and self.last_vad_end == -1 and not self.has_sent_begin and self.mode == self.MODE_ASR:
                    await self._send_event("onSentenceBegin", info="检测到句子开始")
                    self.has_sent_begin = True # 标记已发送

                # ── 中间结果 (Partial result) 处理，仅在 ASR 模式下 ──
                if self.mode == self.MODE_ASR and self.last_vad_beg > -1 and self.last_vad_end == -1:
                    # 计算当前语音段在 audio_vad 中的起始样本位置
                    beg_sample = int((self.last_vad_beg - self.offset) * config.sample_rate / 1000)
                    partial_audio = self.audio_vad[beg_sample:] # 获取从语音开始到当前的音频数据
                    
                    # 仅当音频长度大于一个较小阈值 (如0.3秒) 时才进行中间识别，避免过于频繁或对太短音频进行识别
                    if len(partial_audio) > config.sample_rate * 0.3:
                        # 使用 ASR 模型获取中间识别结果 (use_itn=True 通常用于最终结果，但这里也用，可按需调整)
                        partial_res = asr(partial_audio, self.lang, self.cache_asr, True)
                        if partial_res and partial_res[0]["text"]:
                            partial_text = partial_res[0]["text"] # 格式化识别文本
                            # 发送 onSentenceChange 事件，包含中间识别结果
                            await self._send_event("onSentenceChange", data=format_str_v3(partial_text), info=json.dumps(partial_res[0], ensure_ascii=False))
                            self.last_active_time = time.monotonic() # 更新活动时间，因为有语音活动

                # ── 句子结束 (Sentence end) 处理 ──
                if self.last_vad_beg > -1 and self.last_vad_end > -1: # 如果 VAD 同时检测到了开始和结束
                    # 调整时间戳，减去已处理的偏移量
                    self.last_vad_beg -= self.offset
                    self.last_vad_end -= self.offset
                    self.offset += self.last_vad_end # 更新偏移量为当前段的结束时间
                    
                    # 计算在 audio_vad 中的起止样本位置
                    beg = int(self.last_vad_beg * config.sample_rate / 1000)
                    end = int(self.last_vad_end * config.sample_rate / 1000)

                    audio_seg = self.audio_vad[beg:end] # 提取完整的语音段音频
                    self.audio_vad = self.audio_vad[end:] # 从 audio_vad 中移除已处理的段
                    
                    # 重置 VAD 时间戳和句子开始标记，为下一句做准备
                    self.last_vad_beg = self.last_vad_end = -1
                    self.has_sent_begin = False
                    self.hit = False # 重置说话人验证命中状态
                    
                    # —— 唤醒模式 (Wake mode) 下的处理 ——
                    if self.mode == self.MODE_WAKE:
                        # 运行一次 ASR 获取文本，检查是否包含唤醒词
                        # 注意：这里的 ASR cache 是临时的 {}，不使用流式缓存，因为只为这一段做一次性识别
                        result = asr(audio_seg, self.lang, {}, True) 
                        if result and result[0]["text"]:
                            final_text = result[0]["text"].strip() # 获取识别文本并去除首尾空格
                            if final_text: # 如果识别结果非空
                                
                                text_for_pinyin_conversion = final_text
                                
                                for tag in emoji_dict.keys(): 
                                    text_for_pinyin_conversion = text_for_pinyin_conversion.replace(tag, "")
                                
                                raw_pinyin_list = lazy_pinyin(text_for_pinyin_conversion)
                                
                                filtered_pinyin = [item for item in raw_pinyin_list if item.isalpha()]
                                
                                text_py = "".join(filtered_pinyin).lower()
                                
                                logger.debug(f"唤醒词匹配：原始ASR='{final_text}', 清理后用于拼音='{text_for_pinyin_conversion}', 生成拼音='{text_py}'")  

                                matched_wake_word = None
                                # 遍历所有配置的唤醒词
                                for wake_word in config.wake_words:
                                    # 生成当前唤醒词的拼音（与 WAKE_PINYINS 生成方式一致）
                                    wake_py = "".join(lazy_pinyin(wake_word)).lower()
                                    # 检查是否存在于当前识别结果的拼音中
                                    if wake_py in text_py:
                                        matched_wake_word = wake_word
                                        break  # 找到第一个匹配即可退出
                                
                                # 返回到匹配到的原始唤醒词
                                if matched_wake_word:
                                    logger.info(f"🎉 检测到唤醒词: {matched_wake_word} (原始ASR: {final_text})")
                                    await self._send_event("onWakeWord", data=matched_wake_word, info=f"检测到唤醒词: {matched_wake_word}")
                                    self._switch_to_asr()
                                
                                # 返回的唤醒词是asr识别的唤醒词
                                # if any(wp in text_py for wp in WAKE_PINYINS):
                                #     logger.info(f"🎉 检测到唤醒词 (原始文本: {final_text})") 
                                #     await self._send_event("onWakeWord", data=format_str_v3(final_text), info="检测到唤醒词")
                                #     self._switch_to_asr() 
                                # else:
                                #     logger.debug(f"唤醒模式下未检测到唤醒词，识别文本: {final_text} (拼音: {text_py})")

                    # —— ASR 模式 (ASR mode) 下的处理 ——
                    else: # self.mode == self.MODE_ASR
                        if not (self.sv and not self.hit):
                            result = asr(audio_seg, self.lang, self.cache_asr, True)
                            if result and result[0]["text"]:
                                raw_asr_text = result[0]["text"] 
                                formatted_text_to_send = format_str_v3(raw_asr_text) 
                                
                                text_for_length_check = formatted_text_to_send
                                for char_to_remove in emo_set.union(event_set):
                                    text_for_length_check = text_for_length_check.replace(char_to_remove, "")
                                punctuation_pattern = r"[＂＃＄％＆＇（）＊＋，－／：；＜＝＞＠［＼］＾＿｀｛｜｝～｟｠｢｣､　、〃〈〉《》「」『』【】〔〕〖〗〘〙〚〛〜〝〞〟〰〾〿–—‘’‛“”„‟…‧﹏.!\"#$%&'()*+,-./:;<=>?@[\]^_`{|}~]+"
                                text_for_length_check = re.sub(punctuation_pattern, "", text_for_length_check)
                                text_for_length_check = text_for_length_check.strip()
                                text_content_only = re.sub(r'\s+', '', text_for_length_check) 
                                effective_length = len(text_content_only)
                                
                                logger.debug(
                                    f"ASR 结果评估: 原始格式化='{formatted_text_to_send}', "
                                    f"去表情标点后='{text_for_length_check}', "
                                    f"去所有空格后内容='{text_content_only}', "
                                    f"有效长度={effective_length}"
                                )

                                if effective_length >= 3:
                                    await self._send_event("onSentenceEnd", data=formatted_text_to_send, info=json.dumps(result[0], ensure_ascii=False))
                                    self.last_active_time = time.monotonic() 
                                else:
                                    logger.info(
                                        f"ASR 模式下识别结果 '{formatted_text_to_send}' "
                                        f"(清理后长度 {effective_length}) 少于3个有效字符，已忽略发送。"
                                    )
                                    self.last_active_time = time.monotonic()
                            else: 
                                self.last_active_time = time.monotonic()
                                logger.debug("ASR 模式下，VAD 检测到语音结束，但 ASR 未返回有效文本。")
                        else: 
                            logger.debug("ASR 模式下，需要说话人验证但未通过，未发送 onSentenceEnd。")
                            self.last_active_time = time.monotonic() 

                        self.cache_asr.clear()


        # ───  说话人验证 (Speaker Verification)  ───
        # 条件：ASR 模式，VAD 已检测到语音开始，启用了 SV，本句话尚未验证通过，且已发送句子开始事件
        if (
            self.mode == self.MODE_ASR
            and self.last_vad_beg > -1 # -1 通常表示尚未设置，确保 VAD 已标记语音开始
            and self.sv                 # SV 功能已启用
            and not self.hit            # 本句话尚未验证成功
            and self.has_sent_begin     # 确保是在一个有效的句子内部进行 SV
        ):
            # 计算当前 VAD 语音段在 self.audio_vad 缓冲区中的起始样本索引
            # self.last_vad_beg 是当前语音段开始时间(ms)（相对于流开始，未减去 offset）
            # self.offset 是已处理完毕的语音段的总时长(ms)
            # 因此 (self.last_vad_beg - self.offset) 是当前语音段在 self.audio_vad 中的开始时间(ms)
            start_ms_in_buffer = self.last_vad_beg - self.offset
            beg_sample_sv_in_buffer = int(start_ms_in_buffer * config.sample_rate / 1000)

            # 安全检查：确保计算出的索引起码不为负，并且在 audio_vad 的有效范围内
            if beg_sample_sv_in_buffer < 0:
                # logger.debug(f"SV: Corrected beg_sample_sv_in_buffer from {beg_sample_sv_in_buffer} to 0. last_vad_beg={self.last_vad_beg}, offset={self.offset}")
                beg_sample_sv_in_buffer = 0

            if beg_sample_sv_in_buffer < len(self.audio_vad):
                audio_for_sv = self.audio_vad[beg_sample_sv_in_buffer:]
                
                # 确保有足够时长的音频进行 SV (例如至少 0.5 秒)
                if len(audio_for_sv) >= int(config.sample_rate * 0.5): # 0.5秒的样本数
                    # logger.debug(f"Attempting SV on audio of length {len(audio_for_sv)/config.sample_rate:.2f}s. Start_ms_in_buffer: {start_ms_in_buffer}ms")
                    
                    current_hit_status, speaker_name = self.speaker_verify(audio_for_sv, config.sv_thr)
                    
                    if current_hit_status:
                        # 关键：只有当 SV 结果为 hit 且 self.hit 之前是 False 时，才更新状态并发送事件
                        # 这样可以避免在一句话中多次 SV 成功时重复发送 speakerVerified 事件
                        if not self.hit: # 如果之前未命中，现在命中了
                            self.hit = True # 更新全局的命中状态
                            logger.info(f"🎤 说话人验证成功: {speaker_name}")
                            await self._send_event("speakerVerified", data=speaker_name, info="检测到已注册说话人")
                            self.last_active_time = time.monotonic() # SV 成功也算一次活动

        # ───  无活动超时检测 (Inactivity timeout) ───
        # 条件：ASR 模式，并且距离上次活动时间已超过配置的唤醒超时时间
        if self.mode == self.MODE_ASR and self.enable_wake and \
           (time.monotonic() - self.last_active_time > config.wake_timeout):
            logger.info(f"ASR 模式下无活动超过 {config.wake_timeout} 秒，准备切换回唤醒模式。")
            await self._send_event("onASRTimeout", info=f"ASR模式下超过 {config.wake_timeout} 秒无活动")
            self._switch_to_wake() # 切换回唤醒模式

    # ========================================================
    # 🔒 说话人验证 (Speaker Verification) 
    # ========================================================
    @staticmethod
    def speaker_verify(audio: np.ndarray, sv_thr: float):
        """对给定音频与注册库逐一比对，返回 (是否命中, 命中说话人ID)"""
        hit = False
        hit_spk = ""
        for k, v in reg_spks.items():
            res_sv = sv_pipeline([audio, v["data"]], sv_thr)
            if res_sv["score"] >= sv_thr:
                hit = True
                hit_spk = k
            logger.info(f"[说话人验证] 音频长度: {len(audio)}; 阈值: {sv_thr}; 命中: {hit}; 说话人: {k}; 结果: {res_sv}")
        return hit, hit_spk

# ============================================================
# 🌐 路由
# ============================================================
@app.get("/")
async def index():
    # 提供一个简单的 HTML 客户端页面用于测试 WebSocket 连接
    try:
        with open("client_wss.html", "r", encoding="utf-8") as f:
            return HTMLResponse(f.read())
    except FileNotFoundError:
        logger.error("client_wss.html 文件未找到。请确保该文件存在于服务器运行目录下。")
        return HTMLResponse("<html><body><h1>错误：客户端HTML文件未找到</h1></body></html>", status_code=404)


@app.websocket("/ws/transcribe")
async def websocket_transcribe(websocket: WebSocket):
    # 解析 WebSocket URL 中的查询参数
    qs = parse_qs(websocket.scope["query_string"].decode())
    # 获取 'sv' 参数，默认为 'false'，并转换为布尔值
    sv = qs.get("sv", ["false"])[0].lower() in {"true", "1", "t", "y", "yes", "是"}
    # 获取 'lang' 参数，默认为 'auto'
    lang = qs.get("lang", ["auto"])[0]
    # 获取 'wake' 参数，默认为 'true' (启用唤醒模式)，并转换为布尔值
    wake = qs.get("wake", ["true"])[0].lower() in {"true", "1", "t", "y", "yes", "是"}

    logger.info(f"新的 WebSocket 连接请求: sv={sv}, lang='{lang}', wake={wake}")
    
    # 创建 AudioStreamHandler 实例来处理该 WebSocket 连接
    handler = AudioStreamHandler(websocket, sv=sv, lang=lang, wake=wake)
    await handler.start() # 开始处理音频流

# ============================================================
# ⌨️ CLI 启动
# ============================================================
if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="运行 FastAPI 语音识别 WebSocket 服务 (可选唤醒模式)")
    parser.add_argument("--port", type=int, default=17556, help="服务监听的端口号")
    args = parser.parse_args()

    # 根据配置判断唤醒词模式是否开启
    wake_mode_status = '开启' if config.wake_words and config.wake_timeout > 0 else '关闭'
    if config.wake_words and config.wake_timeout > 0:
        logger.info(f"唤醒词列表: {config.wake_words}")
        logger.info(f"唤醒词拼音 (用于匹配): {WAKE_PINYINS}")
        logger.info(f"ASR无活动超时切换回唤醒模式: {config.wake_timeout} 秒")
    
    logger.info(f"启动服务于 0.0.0.0:{args.port} (唤醒词模式: {wake_mode_status}) …")
    # 使用 uvicorn 启动 FastAPI 应用
    uvicorn.run(app, host="0.0.0.0", port=args.port)